package cn.me.alphamq.nameserver.event.spi.listener.cluster.master;

import cn.me.alphamq.common.entity.SlaveAckEntity;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.event.model.cluster.ReplicationAckEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.store.ReplicationAckManager;
import cn.me.alphamq.common.store.StopWatchManager;

/**
 * 主从复制中主节点的监听器，用于监听从节点复制完成的 ack 事件
 *
 * @author f
 */
public class MasterReplicationAckListener implements Listener<ReplicationAckEvent> {
    @Override
    public void onReceive(ReplicationAckEvent event) throws Exception {
        ReplicationAckManager replicationAckManager = Container.getReplicationAckManager();
        String replicationMsgId = event.getMsgId();

        SlaveAckEntity slaveAckEntity = replicationAckManager.getSlaveAck(replicationMsgId);
        // 如果 ackEntity 是 null，代表这个复制事件已经给 broker 返回响应，不需要再获取 ack 了
        if (slaveAckEntity == null) {
            return;
        }

        // 只有在需要的 ack 数量减到 0 时，进行 移除操作 和 唤醒暂停的任务
        if (slaveAckEntity.getNeedAckCount().decrementAndGet() != 0) {
            return;
        }

        // 移除等待 ack 响应的实体
        replicationAckManager.removeSlaveAck(replicationMsgId);

        // 唤醒阻塞的线程
        StopWatchManager.resumeAndRemove(event.getMsgId());
    }
}
